package edu.wisc.hevea.broker;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.wisc.application.ApplicationPolicy;

import edu.wisc.hevea.constants.ApplicationConstants;
import edu.wisc.hevea.exception.BrokerUnavailableException;
import edu.wisc.hevea.messaging.MessagingServiceClient;
import edu.wisc.hevea.zookeeper.util.ZooKeeperUtil;

public class BrokerManager implements Watcher {

	private static final Logger logger = LoggerFactory
		      .getLogger(BrokerManager.class);
	
	private ConcurrentLinkedQueue<String> brokers;
	private String myDataCenter;
	private String rootNode;
	private short replication;
	private ZooKeeper currentZooKeeper; 
	
	private MessagingServiceClient messagingServiceClient;
		
	public synchronized String getNextBroker()
			throws BrokerUnavailableException {
		if (brokers.size() == 0) {
			throw new BrokerUnavailableException("No broker available!");
		}
		String nextBroker = brokers.remove();
		brokers.add(nextBroker);
		return nextBroker;
	}

	public BrokerManager(ApplicationPolicy appPolicy, String hostName,MessagingServiceClient messagingServiceClient) {
		brokers = new ConcurrentLinkedQueue<String>();
		this.myDataCenter = ApplicationConstants.HOST_DC_MAPPING.get(hostName);
		this.messagingServiceClient = messagingServiceClient;
		rootNode = appPolicy.getApplicationId();
		replication = appPolicy.getReplication();
		boot();
	}

	private void boot() {

		int sessionTimeout = 3000;
		String zooKeeperhostname = ZooKeeperUtil
				.getPreferredZookeeperHostNameByDataCenter(myDataCenter);
		try {
			currentZooKeeper = ZooKeeperUtil.getZooKeeperInstance(
					zooKeeperhostname, sessionTimeout, this);
			
			handleAliveBrokerChanged();
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}
	
	private synchronized void handleAliveBrokerChanged() throws KeeperException, InterruptedException {
		List<String> aliveBrokers = ZooKeeperUtil.getChildren(
				currentZooKeeper, getAliveBrokersPath(), this);
		
		int numUniqueBrokersInSameDC = 0;
		Set<String> uniqueBrokersInSameDC = new HashSet<String>();
		for(String currentPreferredBroker : brokers) {
			if(ApplicationConstants.HOST_DC_MAPPING.get(currentPreferredBroker).equals(myDataCenter)) {
				uniqueBrokersInSameDC.add(currentPreferredBroker);
			}
		}
		numUniqueBrokersInSameDC = uniqueBrokersInSameDC.size();
		
		if(numUniqueBrokersInSameDC >= replication && aliveBrokers.containsAll(brokers)) {
		 // No need to change the current broker list
		} else {
			
			for(String currentBroker : brokers) {
				if(!aliveBrokers.contains(currentBroker)) {
					messagingServiceClient.removeServiceInterface(currentBroker);
				}
			}
			
			brokers.clear();
			for(String broker : aliveBrokers) {
				if (ApplicationConstants.HOST_DC_MAPPING.get(broker)
								.equals(myDataCenter)) {
					brokers.add(broker);
				}
			}
			if (brokers.size() < replication && aliveBrokers.size() > 0) {
				while(brokers.size() < replication) {
					brokers.add(getRandomBroker(aliveBrokers));	
				}				
			}
		}
		
		logger.info(" PreferredBroker queue ");
		for(String preferredBroker : brokers) {
			logger.info(preferredBroker);
		}
		
	}
	
	private String getRandomBroker(List<String> aliveBrokers) {
		if(aliveBrokers.size() == 0) {
			return "";
		}
		int index = (int) (Math.random() * aliveBrokers.size());
		logger.info("Returning random broker : " + aliveBrokers.get(index));
		return aliveBrokers.get(index);
	}

	private String getAliveBrokersPath() {
		return "/" + rootNode + "/" + ApplicationConstants.ALIVE_BROKERS;
	}

	@Override
	public void process(WatchedEvent event) {
		
		if(event.getPath() == null) {
			return;
		}

		if (event.getPath().equals(getAliveBrokersPath())) {
			try {
				handleAliveBrokerChanged();
			} catch (KeeperException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

}
